package io.g740.handler;

import io.g740.dto.Column;
import io.g740.dto.FieldMappingDto;
import io.g740.dto.FieldMappingResultDto;
import io.g740.service.MigrationConfig;
import io.g740.util.Constants;
import io.g740.util.SqlUtil;
import io.g740.util.UrlUtil;
import org.apache.ibatis.jdbc.SqlRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 数据迁移处理器
 * <p>
 * 全量保留
 *
 * @author YunKaiLuo
 * @date 2024/10/22
 */
public class FullQuantityReservedDataMigrationHandler implements DataMigrationHandler {

    private static final Logger log = LoggerFactory.getLogger(FullQuantityReservedDataMigrationHandler.class);

    @Override
    public boolean migrate(MigrationConfig migrationConfig, Connection sourceConn, Connection targetConn) throws SQLException {
        // 同一个数据库实例，进行语句复制
        Map<String/*源表字段名，全部小写*/, FieldMappingResultDto> fieldMappingMap = migrationConfig.getFieldMappingDtos().stream()
                .map(fieldMappingDto -> new FieldMappingResultDto(fieldMappingDto.getSourceColumnName()
                        , fieldMappingDto.getTargetColumnName()))
                .collect(Collectors.toMap(FieldMappingDto::getSourceColumnName
                        , Function.identity()));
        Map<String, Column> sourceColumnsInfo = SqlUtil.getColumnsInfo(sourceConn, migrationConfig.getSourceTableInfo().getName());
        if (migrationConfig.isDefaultMigrationOfAllFields()) {
            for (Map.Entry<String, Column> entry : sourceColumnsInfo.entrySet()) {
                fieldMappingMap.putIfAbsent(entry.getKey(), new FieldMappingResultDto(entry.getKey(), entry.getKey()));
            }
        }
        for (Map.Entry<String/*源表字段名，全部小写*/, FieldMappingResultDto> entry : fieldMappingMap.entrySet()) {
            Column column = sourceColumnsInfo.get(entry.getKey());
            if (column != null) {
                FieldMappingResultDto value = entry.getValue();
                value.setSourcePosition(column.getOrdinalPosition());
                value.setSourceTypeName(column.getTypeName());
            }
        }

        Map<String, Column> targetColumnsInfo = SqlUtil.getColumnsInfo(sourceConn, migrationConfig.getTargetTableInfo().getName());
        for (Map.Entry<String/*源表字段名，全部小写*/, FieldMappingResultDto> entry : fieldMappingMap.entrySet()) {
            Column column = targetColumnsInfo.get(entry.getValue().getTargetColumnName());
            if (column != null) {
                FieldMappingResultDto value = entry.getValue();
                value.setTargetTypeName(column.getTypeName());
            }
        }

        // 将源表中的数据全部新增到新表中
        if (UrlUtil.checkIfTheHostAndPortOfMysqlAreEqual(
                migrationConfig.getSourceTableInfo().getDataSourceConfig().getUrl(),
                migrationConfig.getTargetTableInfo().getDataSourceConfig().getUrl()
        )) {
            String replaceInToTable = Constants.getReplaceInToTable(
                    sourceConn.getCatalog(),
                    migrationConfig.getSourceTableInfo().getName(),
                    targetConn.getCatalog(),
                    migrationConfig.getTargetTableInfo().getName(),
                    fieldMappingMap.values()
            );
            log.debug("【数据迁移】插入语句 {}", replaceInToTable);
            SqlUtil.execute(targetConn, replaceInToTable);
        } else {
            /**
             * 利用sqlRunner进行插入
             */
            String selectSourceTable = Constants.getSelectTable(sourceConn.getCatalog(),
                    migrationConfig.getSourceTableInfo().getName(), fieldMappingMap.keySet());
            SqlRunner sqlRunner = new SqlRunner(sourceConn);
            SqlRunner targetSqlRunner = new SqlRunner(targetConn);
            List<Map<String, Object>> result = sqlRunner.selectAll(selectSourceTable);
            Collection<FieldMappingResultDto> values = fieldMappingMap.values();

            String replaceInToTableWithPlaceholders = Constants.getReplaceInToTableWithPlaceholders(
                    sourceConn.getCatalog(),
                    migrationConfig.getSourceTableInfo().getName(),
                    targetConn.getCatalog(),
                    migrationConfig.getTargetTableInfo().getName(),
                    values
            );
            targetConn.setAutoCommit(false);
            // 500条数据提交一次
            final int batchSize = 500;
            int count = 0;
            int commitCount = 0;
            // 这里的效率有点慢
            for (Map<String, Object> row : result) {
                List<Object> record = new LinkedList<>();
                for (FieldMappingResultDto filed : values) {
                    Object colValue = row.get(filed.getSourceColumnName().toUpperCase(Locale.ENGLISH));
                    record.add(colValue);
                }
                targetSqlRunner.insert(replaceInToTableWithPlaceholders, record.toArray());
                if (batchSize == ++count) {
                    targetConn.commit();
                    count = 0;
                    log.debug("【数据迁移】批量提交，轮数：{}", commitCount);
                }
            }
            targetConn.commit();
        }
        return false;
    }
}
